package cn.linjianhui.flink.sample;

import cn.linjianhui.flink.sample.model.OmsOrderDetail;
import cn.linjianhui.flink.sample.util.ParamUtils;
import cn.linjianhui.flink.sample.util.Utils;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.Properties;

public class TopNProductByCategoryJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = Utils.getStreamExecutionEnvironment();

        ParameterTool tool = ParameterTool.fromArgs(args);
        String category = tool.get("category", StringUtils.EMPTY).trim();
        if(StringUtils.isBlank(category)) {
            throw new RuntimeException("category param cannot be null");
        }

        Properties props = ParamUtils.getKafkaConsumerProperty(args, String.format("topN-product-%s", category), "earliest");

        DataStream<String> sourceStream = env.addSource(new FlinkKafkaConsumer011<>("OMS-Order-Ware", new SimpleStringSchema(), props));

        DataStream<Tuple2<String, Long>> orderDetailStream = sourceStream
                .map((MapFunction<String, OmsOrderDetail>) s -> JSON.parseObject(s, OmsOrderDetail.class))
                .filter((FilterFunction<OmsOrderDetail>) detail -> {
                    long currentDayTimestamp = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
                    return detail.getPAYMENTDATE() >= currentDayTimestamp && category.equals(detail.getVirtualCategory());
                })
                .map((MapFunction<OmsOrderDetail, Tuple2<String, Long>>) detail -> new Tuple2<>(detail.getWARECODE(), Long.valueOf(detail.getQTY())))
                .returns(TypeInformation.of(new TypeHint<Tuple2<String, Long>>() { }));

        DataStream<Tuple2<String, Long>> resultStream = orderDetailStream
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
                .sum(1);

        FlinkJedisPoolConfig jedisPoolConfig = ParamUtils.getJedisPoolConfig(args);
        resultStream.addSink(new RedisSink<>(jedisPoolConfig, new RedisMapper<Tuple2<String, Long>>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.ZADD, String.format("flink-sample-sales_%s", category));
            }

            @Override
            public String getKeyFromData(Tuple2<String, Long> output) {
                return output.f0;
            }

            @Override
            public String getValueFromData(Tuple2<String, Long> output) {
                return String.valueOf(output.f1);
            }
        }));

        env.execute(String.format("%s Product TopN Job", category));
    }
}
